package org.hsqldb.index;

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.hsqldb.HsqlNameManager.HsqlName;
import org.hsqldb.Constraint;
import org.hsqldb.OpTypes;
import org.hsqldb.Row;
import org.hsqldb.RowBTree;
import org.hsqldb.SchemaObject;
import org.hsqldb.Session;
import org.hsqldb.Table;
import org.hsqldb.TableBase;
import org.hsqldb.Tokens;
import org.hsqldb.TransactionManager;
import org.hsqldb.error.Error;
import org.hsqldb.error.ErrorCode;
import org.hsqldb.lib.ArrayUtil;
import org.hsqldb.lib.OrderedHashSet;
import org.hsqldb.lib.ReadWriteLockDummy;
import org.hsqldb.navigator.RowIterator;
import org.hsqldb.persist.CachedObject;
import org.hsqldb.persist.PersistentStore;
import org.hsqldb.persist.RowStoreBTree;
import org.hsqldb.rights.Grantee;
import org.hsqldb.types.Type;

import sun.net.www.content.audio.x_aiff;

public class IndexBTree implements Index{

	// fields
    private final long       persistenceId;
    protected final HsqlName name;
    private final boolean[]  colCheck;
    final int[]              colIndex;
    private final int[]      defaultColMap;
    final Type[]             colTypes;
    private final boolean[]  colDesc;
    private final boolean[]  nullsLast;
    final boolean            isSimpleOrder;
    final boolean            isSimple;
    protected final boolean  isPK;        // PK with or without columns
    protected final boolean  isUnique;    // DDL uniqueness
    protected final boolean  isConstraint;
    private final boolean    isForward;
    private boolean          isClustered;
    private static final IndexRowIterator emptyIterator =
        new IndexRowIterator(null, (PersistentStore) null, null, null, 0,
                             false, false);
    protected TableBase table;
    int                 position;
    
    //
    Object[] nullData;

    //
    ReadWriteLock lock;
    Lock          readLock;
    Lock          writeLock;

    /**
     * Constructor declaration
     *
     * @param name HsqlName of the index
     * @param id persistnece id
     * @param table table of the index
     * @param columns array of column indexes
     * @param descending boolean[]
     * @param nullsLast boolean[]
     * @param colTypes array of column types
     * @param pk if index is for a primary key
     * @param unique is this a unique index
     * @param constraint does this index belonging to a constraint
     * @param forward is this an auto-index for an FK that refers to a table
     *   defined after this table
     */
    public IndexBTree(HsqlName name, long id, TableBase table, int[] columns,
                    boolean[] descending, boolean[] nullsLast,
                    Type[] colTypes, boolean pk, boolean unique,
                    boolean constraint, boolean forward) {

        this.persistenceId = id;
        this.name          = name;
        this.colIndex      = columns;
        this.colTypes      = colTypes;
        this.colDesc       = descending == null ? new boolean[columns.length]
                                                : descending;
        this.nullsLast     = nullsLast == null ? new boolean[columns.length]
                                               : nullsLast;
        this.isPK          = pk;
        this.isUnique      = unique;
        this.isConstraint  = constraint;
        this.isForward     = forward;
        this.table         = table;
        this.colCheck      = table.getNewColumnCheckList();

        ArrayUtil.intIndexesToBooleanArray(colIndex, colCheck);

        this.defaultColMap = new int[columns.length];

        ArrayUtil.fillSequence(defaultColMap);

        boolean simpleOrder = colIndex.length > 0;

        for (int i = 0; i < colDesc.length; i++) {
            if (this.colDesc[i] || this.nullsLast[i]) {
                simpleOrder = false;
            }
        }

        isSimpleOrder = simpleOrder;
        isSimple      = isSimpleOrder && colIndex.length == 1;
        nullData      = new Object[colIndex.length];

        //
        switch (table.getTableType()) {

            case TableBase.MEMORY_TABLE :
            case TableBase.CACHED_TABLE :
            case TableBase.TEXT_TABLE :
                lock = new ReentrantReadWriteLock();
                break;

            default :
                lock = new ReadWriteLockDummy();
                break;
        }
        readLock  = lock.readLock();
        writeLock = lock.writeLock();
    }
    /**
     * SchemaObjecte interface
     */
    public int getType() {
        return SchemaObject.INDEX;
    }

	public HsqlName getName() {
        return name;
    }

    public HsqlName getCatalogName() {
        return name.schema.schema;
    }

    public HsqlName getSchemaName() {
        return name.schema;
    }

    public Grantee getOwner() {
        return name.schema.owner;
    }

    public OrderedHashSet getReferences() {
        return new OrderedHashSet();
    }

    public OrderedHashSet getComponents() {
        return null;
    }

    public void compile(Session session, SchemaObject parentObject) {}

    public String getSQL() {

        StringBuffer sb = new StringBuffer();

        sb = new StringBuffer(64);

        sb.append(Tokens.T_CREATE).append(' ');

        if (isUnique()) {
            sb.append(Tokens.T_UNIQUE).append(' ');
        }

        sb.append(Tokens.T_INDEX).append(' ');
        sb.append(getName().statementName);
        sb.append(' ').append(Tokens.T_ON).append(' ');
        sb.append(((Table) table).getName().getSchemaQualifiedStatementName());

        int[] col = getColumns();
        int   len = getVisibleColumns();

        sb.append(((Table) table).getColumnListSQL(col, len));

        return sb.toString();
    }

    public long getChangeTimestamp() {
        return 0;
    }


	/**
	 *  IndexInterface
	 */
    public RowIterator emptyIterator() {
        return emptyIterator;
    }

    public int getPosition() {
        return position;
    }

    public void setPosition(int position) {
        this.position = position;
    }

    public long getPersistenceId() {
        return persistenceId;
    }

    /**
     * Returns the count of visible columns used
     */
    public int getVisibleColumns() {
        return colIndex.length;
    }

    /**
     * Returns the count of visible columns used
     */
    public int getColumnCount() {
        return colIndex.length;
    }

    /**
     * Is this a UNIQUE index?
     */
    public boolean isUnique() {
        return isUnique;
    }

    /**
     * Does this index belong to a constraint?
     */
    public boolean isConstraint() {
        return isConstraint;
    }

    /**
     * Returns the array containing column indexes for index
     */
    public int[] getColumns() {
        return colIndex;
    }

    /**
     * Returns the array containing column indexes for index
     */
    public Type[] getColumnTypes() {
        return colTypes;
    }

    public boolean[] getColumnDesc() {
        return colDesc;
    }

    public int[] getDefaultColumnMap() {
        return this.defaultColMap;
    }

    /**
     * Returns a value indicating the order of different types of index in
     * the list of indexes for a table. The position of the groups of Indexes
     * in the list in ascending order is as follows:
     *
     * primary key index
     * unique constraint indexes
     * autogenerated foreign key indexes for FK's that reference this table or
     *  tables created before this table
     * user created indexes (CREATE INDEX)
     * autogenerated foreign key indexes for FK's that reference tables created
     *  after this table
     *
     * Among a group of indexes, the order is based on the order of creation
     * of the index.
     *
     * @return ordinal value
     */
    public int getIndexOrderValue() {

        if (isPK) {
            return 0;
        }

        if (isConstraint) {
            return isForward ? 4
                             : isUnique ? 0
                                        : 1;
        } else {
            return 2;
        }
    }

    public boolean isForward() {
        return isForward;
    }

    public void setTable(TableBase table) {
        this.table = table;
    }

    public void setClustered(boolean clustered) {
        isClustered = clustered;
    }

    public boolean isClustered() {
        return isClustered;
    }

    /**
     * Returns the node count.
     */
    public int size(Session session, PersistentStore store) {

        readLock.lock();

        try {
            return store.elementCount(session);
        } finally {
            readLock.unlock();
        }
    }

    public int sizeUnique(PersistentStore store) {

        readLock.lock();

        try {
            return store.elementCountUnique(this);
        } finally {
            readLock.unlock();
        }
    }

    public boolean isEmpty(PersistentStore store) {

        readLock.lock();

        try {
        	return getAccessor(store) == null;
        } finally {
            readLock.unlock();
        }
    }
    
    NodeBTree getAccessor(PersistentStore store) {
        return (NodeBTree)store.getAccessor(this);
    }

    public void checkIndex(PersistentStore store) {
        //TODO:
    	
    }


    /**
     * Compares two table rows based on the columns of this index. The rowColMap
     * parameter specifies which columns of the other table are to be compared
     * with the colIndex columns of this index. The rowColMap can cover all or
     * only some columns of this index.
     *
     * @param session Session
     * @param a row from another table
     * @param rowColMap column indexes in the other table
     * @param b a full row in this table
     * @return comparison result, -1,0,+1
     */
    public int compareRowNonUnique(Session session, Object[] a, Object[] b,
                                   int[] rowColMap) {

        int fieldcount = rowColMap.length;

        for (int j = 0; j < fieldcount; j++) {
            int i = colTypes[j].compare(session, a[colIndex[j]],
                                        b[rowColMap[j]]);

            if (i != 0) {
                return i;
            }
        }

        return 0;
    }

    public int compareRowNonUnique(Session session, Object[] a, Object[] b,
                                   int[] rowColMap, int fieldCount) {

        for (int j = 0; j < fieldCount; j++) {
            int i = colTypes[j].compare(session, a[colIndex[j]],
                                        b[rowColMap[j]]);

            if (i != 0) {
                return i;
            }
        }

        return 0;
    }

    /**
     * As above but use the index column data
     */
    public int compareRowNonUnique(Session session, Object[] a, Object[] b,
                                   int fieldCount) {

        for (int j = 0; j < fieldCount; j++) {
            int i = colTypes[j].compare(session, a[colIndex[j]],
                                        b[colIndex[j]]);

            if (i != 0) {
                return i;
            }
        }

        return 0;
    }

    public int compareRow(Session session, Object[] a, Object[] b) {

        for (int j = 0; j < colIndex.length; j++) {
            int i = colTypes[j].compare(session, a[colIndex[j]],
                                        b[colIndex[j]]);

            if (i != 0) {
                if (isSimpleOrder) {
                    return i;
                }

                boolean nulls = a[colIndex[j]] == null
                                || b[colIndex[j]] == null;

                if (colDesc[j] && !nulls) {
                    i = -i;
                }

                if (nullsLast[j] && nulls) {
                    i = -i;
                }

                return i;
            }
        }

        return 0;
    }
    
    boolean hasNulls(Session session, Object[] rowData) {

        if (colIndex.length == 1) {
            return rowData[colIndex[0]] == null;
        }

        boolean normal = session == null ? true
                                         : session.database.sqlUniqueNulls;

        for (int j = 0; j < colIndex.length; j++) {
            if (rowData[colIndex[j]] == null) {
                if (normal) {
                    return true;
                }
            } else {
                if (!normal) {
                    return false;
                }
            }
        }

        return !normal;
    }
 

	@Override
	public boolean existsParent(Session session, PersistentStore store,
			Object[] rowdata, int[] rowColMap) {
		NodeBTree node = (NodeBTree)findNode(session, store, rowdata, rowColMap,
                rowColMap.length, OpTypes.EQUAL,
                TransactionManager.ACTION_REF, false);
		return node != null;
	}

	/**
     * Return the first node equal to the indexdata object. The rowdata has the
     * same column mapping as this index.
     *
     * @param session session object
     * @param store store object
     * @param rowdata array containing index column data
     * @param matchCount count of columns to match
     * @param compareType int
     * @param reversed boolean
     * @param map boolean[]
     * @return iterator
     */
	public RowIterator findFirstRow(Session session, PersistentStore store,
			Object[] rowdata, int matchCount, int distinctCount,
			int compareType, boolean reversed, boolean[] map) {
		if (compareType == OpTypes.MAX) {
            return lastRow(session, store);
        }

        NodeBTreeData node = (NodeBTreeData)findNode(session, store, rowdata, defaultColMap,
                                matchCount, compareType,
                                TransactionManager.ACTION_READ, reversed);

        if (node == null) {
            return emptyIterator;
        }
        return new IndexRowIterator(session, store, this, node, distinctCount,
                                    false, reversed);
	}

	@Override
	public RowIterator findFirstRow(Session session, PersistentStore store,
			Object[] rowdata) {
		NodeBTreeData node = (NodeBTreeData)findNode(session, store, rowdata, colIndex,
                colIndex.length, OpTypes.EQUAL,
                TransactionManager.ACTION_READ, false);

		if (node == null) {
			return emptyIterator;
		}
		
		return new IndexRowIterator(session, store, this, node, 0, false,
		                    false);
	}

	@Override
	public RowIterator findFirstRow(Session session, PersistentStore store,
			Object[] rowdata, int[] rowColMap) {
		NodeBTreeData node = (NodeBTreeData)findNode(session, store, rowdata, rowColMap,
                rowColMap.length, OpTypes.EQUAL,
                TransactionManager.ACTION_READ, false);
		
		if (node == null) {
			return emptyIterator;
		}
		
		return new IndexRowIterator(session, store, this, node, 0, false,
		                    false);
	}

	@Override
	public RowIterator findFirstRowNotNull(Session session,
			PersistentStore store) {
		NodeBTreeData node = (NodeBTreeData)findNode(session, store, nullData, this.defaultColMap,
                1, OpTypes.NOT,
                TransactionManager.ACTION_READ, false);

		if (node == null) {
			return emptyIterator;
		}
		
		return new IndexRowIterator(session, store, this, node, 0, false,
		                    false);
	}

	@Override
	public RowIterator firstRow(PersistentStore store) {
		readLock.lock();

        try {
            NodeBTreeBranch x = (NodeBTreeBranch)getAccessor(store);

            if (x == null) {
                return emptyIterator;
            }
            NodeBTreeBranch node=x;
            while(true){
            	if(node.isLeaf){
            		break;
            	}
            	node=(NodeBTreeBranch)node.getChild(store, 0);
            }
            
            return new IndexRowIterator(null, store, this, (NodeBTreeData)node.getChild(store, 0), 0, false, false);
            
        } finally {
            readLock.unlock();
        }
	}

	@Override
	public RowIterator firstRow(Session session, PersistentStore store) {
		readLock.lock();

        try {
            NodeBTreeBranch x = (NodeBTreeBranch)getAccessor(store);
            if(x == null)
            	return emptyIterator;
            while(true){
            	if(x.isLeaf){
            		break;
            	}
            	x=(NodeBTreeBranch)x.getChild(store, 0);
            }
            NodeBTreeData data=(NodeBTreeData)x.getChild(store, 0);
            
            while (session != null && data != null) {
                Row row = data.getRow(store);
            	//Row row=null;
                if (session.database.txManager.canRead(
                        session, row, TransactionManager.ACTION_READ, null)) {
                    break;
                }
                data = (NodeBTreeData)next(store, data);
                //data=data.nodeNext;
                //x = (NodeBTree)next(store, x);
            }

            if (data == null) {
                return emptyIterator;
            }

            return new IndexRowIterator(session, store, this, data, 0, false,
                                        false);
        } finally {
            readLock.unlock();
        }
	}

	@Override
	public RowIterator lastRow(Session session, PersistentStore store) {
		readLock.lock();

        try {
            NodeBTreeBranch x = (NodeBTreeBranch)getAccessor(store);

            if (x == null) {
                return emptyIterator;
            }
            while(true){
            	if(x.isLeaf)
            		break;
            	x=(NodeBTreeBranch)x.getChild(store, x.getChildNumber()-1);
            }
            NodeBTreeData l = (NodeBTreeData)x.getChild(store, x.getChildNumber()-1);
            return new IndexRowIterator(session, store, this, l, 0, false,
                        true);
            
        } finally {
            readLock.unlock();
        }
	}

	
	

	/**
     * Compare two rows of the table for inserting rows into unique indexes
     * Supports descending columns.
     *
     * @param session Session
     * @param newRow data
     * @param existingRow data
     * @param useRowId boolean
     * @param start int
     * @return comparison result, -1,0,+1
     */
    int compareRowForInsertOrDelete(Session session, Row newRow,
                                    Row existingRow, boolean useRowId,
                                    int start) {

        Object[] a = newRow.getData();
        Object[] b = existingRow.getData();

        for (int j = start; j < colIndex.length; j++) {
            int i = colTypes[j].compare(session, a[colIndex[j]],
                                        b[colIndex[j]]);

            if (i != 0) {
                if (isSimpleOrder) {
                    return i;
                }

                boolean nulls = a[colIndex[j]] == null
                                || b[colIndex[j]] == null;

                if (colDesc[j] && !nulls) {
                    i = -i;
                }

                if (nullsLast[j] && nulls) {
                    i = -i;
                }

                return i;
            }
        }

        if (useRowId) {
            return newRow.getPos() - existingRow.getPos();
        }

        return 0;
    }
    int compareObject(Session session, Object[] a, Object[] b,int[] rowColMap, int position) {
    	return colTypes[position].compare(session, a[colIndex[position]],
	                                b[rowColMap[position]]);
	}
    
    CachedObject next(PersistentStore store, NodeBTreeData x) {
    	if (x == null) {
            return null;
        }
    	
    	NodeBTreeData data = x;
		if (data.nodeNext != null) {
			return data.nodeNext;
		} else {
			NodeBTreeBranch node = (NodeBTreeBranch)data.getParent(store);
			if(node == null)
				return null;
			int pos = data.getPositionInParent();
			if (pos < node.getChildNumber()-1){
				NodeBTreeData d = (NodeBTreeData)node.getChild(store, pos+1);
				return d;
			}
			NodeBTreeBranch nextLeaf = (NodeBTreeBranch) node.getNextLeaf(store);
			if(nextLeaf == null)
				return null;
			return nextLeaf.getChild(store, 0);
			/*
			while (pos == node.getChildNumber()-1) {
				pos = node.getPositionInParent();
				node = (NodeBTreeBranch)node.getParent();
				if (node == null){
					return null;
				}
			}
			node = (NodeBTreeBranch)node.getChild(store, pos+1);
			while (!node.isLeaf){
				node = (NodeBTreeBranch)node.getChild(store, 0);
			}
			NodeBTreeData d = (NodeBTreeData)node.getChild(store, 0);
			return d;
			*/
		}
    }

    CachedObject last(PersistentStore store, NodeBTreeData x) {

        if (x == null) {
            return null;
        }
        
		NodeBTreeData data = x;
		if (data.nodePrev != null) {
			return data.nodePrev;
		} else {
			NodeBTreeBranch node = (NodeBTreeBranch)data.getParent(store);
			int pos = data.getPositionInParent();
			if (pos > 0){
				NodeBTreeData d = (NodeBTreeData)node.getChild(store, pos-1);
				while (d.nodeNext != null){
					d = d.nodeNext;
				}
				return d;
			}
			while (pos == 0){
				pos = node.getPositionInParent();
				node = (NodeBTreeBranch)x.getParent(store);
				if (node == null){
					return null;
				}
			}
			node = (NodeBTreeBranch)node.getChild(store, pos-1);
			while (!node.isLeaf){
				node = (NodeBTreeBranch)node.getChild(store, node.getChildNumber()-1);
			}
			NodeBTreeData d = (NodeBTreeData)node.getChild(store, node.getChildNumber()-1);
			while (d.nodeNext != null){
				d = d.nodeNext;
			}
			return d;
		}
    }
	private CachedObject last(Session session, PersistentStore store, 
								NodeBTreeData x, int distinctCount){
		if (x == null){
			return null;
		}
		
		while (true) {
            if (distinctCount == 0) {
                x = (NodeBTreeData)last(store, x);
            } else {
                Object[] baseData = x.getData(store);

                return findNode(session, store, baseData, colIndex,
                                distinctCount, OpTypes.SMALLER,
                                TransactionManager.ACTION_READ, false);
            }

            if (x == null) {
                return x;
            }

            if (session == null) {
                return x;
            }

            Row row = x.getRow(store);

            if (session.database.txManager.canRead(
                    session, row, TransactionManager.ACTION_READ, null)) {
                return x;
            }
        }
	}
	private CachedObject next(Session session, PersistentStore store, 
								NodeBTreeData x, int distinctCount){
		if (x == null) {
            return null;
        }

        while (true) {
            if (distinctCount == 0) {
                x = (NodeBTreeData)next(store, x);
            } else {
                Object[] baseData = x.getData(store);

                return findNode(session, store, baseData, colIndex,
                                distinctCount, OpTypes.GREATER,
                                TransactionManager.ACTION_READ, false);
            }

            if (x == null) {
                return x;
            }

            if (session == null) {
                return x;
            }

            Row row = x.getRow(store);

            if (session.database.txManager.canRead(
                    session, row, TransactionManager.ACTION_READ, null)) {
                return x;
            }
        }
	}
    public static class IndexRowIterator implements RowIterator{

		final Session         session;
	    final PersistentStore store;
	    final IndexBTree        index;
	    NodeBTreeData          nextnode;
	    Row                   lastrow;
	    int                   distinctCount;
	    boolean               single;
	    boolean               reversed;

	    /**
	     * When session == null, rows from all sessions are returned
	     */
	    public IndexRowIterator(Session session, PersistentStore store,
	                            IndexBTree index, NodeBTreeData node,
	                            int distinctCount, boolean single,
	                            boolean reversed) {
	        this.session       = session;
	        this.store         = store;
	        this.index         = index;
	        this.distinctCount = distinctCount;
	        this.single        = single;
	        this.reversed      = reversed;

	        if (index == null) {
	            return;
	        }

	        nextnode = node;
	    }

	    public boolean hasNext() {
	        return nextnode != null;
	    }

	    public Row getNextRow() {

	        if (nextnode == null) {
	            release();

	            return null;
	        }

	        NodeBTreeData lastnode = nextnode;

	        if (single) {
	            nextnode = null;
	        } else {
	            index.readLock.lock();
	            store.writeLock();

	            try {
	                if (reversed) {
	                    nextnode = (NodeBTreeData)index.last(session, store, nextnode,
	                                          distinctCount);
	                } else {
	                    nextnode = (NodeBTreeData)index.next(session, store, nextnode,
	                                          distinctCount);
	                }
	            } finally {
	                store.writeUnlock();
	                index.readLock.unlock();
	            }
	        }

	        lastrow = lastnode.getRow(store);
	        return lastrow;
	    }

	    public Object[] getNext() {

	        Row row = getNextRow();

	        return row == null ? null
	                           : row.getData();
	    }

	    public void remove() {
	        store.delete(session, lastrow);
	        store.remove(lastrow.getPos());
	    }

	    public void release() {}

	    public boolean setRowColumns(boolean[] columns) {
	        return false;
	    }

	    public long getRowId() {
	        return nextnode.getPos();
	    }
		
	}
	public int getNodeCount(Session session, RowStoreBTree rowStoreBTree) {
		// TODO Auto-generated method stub
		int count = 0;

        readLock.lock();

        try {
            RowIterator it = firstRow(session, rowStoreBTree);

            while (it.hasNext()) {
                it.getNextRow();

                count++;
            }

            return count;
        } finally {
            readLock.unlock();
        }
	}
	/**
     * Compare two keys of the data for inserting rows into unique indexes
     * Supports descending columns.
     *
     * @param session Session
     * @param newRow data
     * @param existingKeys data
     * @return comparison result, -1,0,+1
     */
	public int compareKeys(Session session, Row newRow, Object[] exisitingKeys){
		Object[] keys = newRow.getData();
		int pos;
		for (pos = 0; pos < exisitingKeys.length; pos++){
			int i = colTypes[pos].compare(session, keys[colIndex[pos]], 
											exisitingKeys[pos]);
			if (i != 0){
				if (isSimpleOrder) {
                    return i;
                }

                boolean nulls = keys[colIndex[pos]] == null
                                || exisitingKeys[pos] == null;

                if (colDesc[pos] && !nulls) {
                    i = -i;
                }

                if (nullsLast[pos] && nulls) {
                    i = -i;
                }

                return i;
			}
		}
		
		return 0;
	}
	public int compareKeys(Session session, Object[] keys, Object[] exisitingKeys){
		int pos;
		for (pos = 0; pos < exisitingKeys.length; pos++){
			int i = colTypes[pos].compare(session, keys[colIndex[pos]], 
											exisitingKeys[pos]);
			if (i != 0){
				if (isSimpleOrder) {
                    return i;
                }

                boolean nulls = keys[colIndex[pos]] == null
                                || exisitingKeys[pos] == null;

                if (colDesc[pos] && !nulls) {
                    i = -i;
                }

                if (nullsLast[pos] && nulls) {
                    i = -i;
                }

                return i;
			}
		}
		
		return 0;
	}
	public void insert(Session session, PersistentStore store, Row row){
		NodeBTreeBranch n;
        int     compare      = -1;
        boolean compareRowId = !isUnique || hasNulls(session, row.getData());
        
        writeLock.lock();
        store.writeLock();
        
        try {
			// get keys
			Object[] keys = row.getData();
			Object[] newKeys = new Object[colTypes.length];

			for (int k = 0; k < colTypes.length; k++){
				newKeys[k] = keys[colIndex[k]];
			}
			// get root
			n = (NodeBTreeBranch)getAccessor(store);
			
			// no root
			if (n == null){
				n = new NodeBTreeBranch();
				n.insertDataAtLeaf(store,newKeys, ((RowBTree)row).getNode(position), 0);
				store.setAccessor(this, n);
				return;
			}
			
			// search in tree
			while (true){
				int len = n.getChildNumber();
				int pos = 0;
				for (pos = 0; pos < len; pos++){
			        if (compareRowId) {
			            compare=row.getPos()-n.getChildPos(pos);
			        }
			        else{
			        	compare = compareKeys(session, row, n.getKeys(pos));
			        }
					if (compare <= 0){
						break;
					}
				}
				if (n.isLeaf){
					// already exists
					if (compare == 0){
						n.insertDulplicateData(store,((RowBTree)row).getNode(position), pos);
					} else{
						n.insertDataAtLeaf(store,newKeys, ((RowBTree)row).getNode(position), pos);
					}
					
					break;
				}
				n = (NodeBTreeBranch)n.getChild(store, pos);
			}
			
			// check child number
			while (n.isOver()){
				NodeBTreeBranch parentBranch = (NodeBTreeBranch)n.getParent(store);
				
				NodeBTreeBranch anotherBranch;
				if(n.isLeaf())
					anotherBranch = n.split(store,n.keyNumber/2);
				else {
					anotherBranch = n.split(store,n.getChildNumber() / 2 + 1);
				}
				// is root
				if (parentBranch == null){
					NodeBTreeBranch rootBranch = new NodeBTreeBranch();
					rootBranch.setIsLeaf(false);
					rootBranch.setChild(store,n, 0);
					n.setParent(store,rootBranch,0);
					rootBranch.setChild(store,anotherBranch, 1);
					anotherBranch.setParent(store,rootBranch,1);
					n.setNextLeaf(store, anotherBranch);
					anotherBranch.setPreLeaf(store, n);
					if(n.isLeaf()){
						rootBranch.setKeys(anotherBranch.getKeys(0), 0);
					}
					else {
						rootBranch.setKeys(n.getKeys(n.getChildNumber() - 1), 0);
						//FIXME:This should be completed by a method
						n.setChild(store,null, n.keyNumber);
						n.setKeys(null, n.keyNumber - 1);
						n.keyNumber--;
					}
					rootBranch.setIsLeaf(false);
					store.setAccessor(this, rootBranch);
					break;
				}
				// normal node
				if (n.isLeaf()){
					anotherBranch.setNextLeaf(store, n.getNextLeaf(store));
					anotherBranch.setPreLeaf(store, n);
					((NodeBTreeBranch)(n.getNextLeaf(store))).setPreLeaf(store, anotherBranch);
					n.setNextLeaf(store, anotherBranch);
					parentBranch.insertIntoNonLeaf(store,anotherBranch.getKeys(0), 
							n, n.getPositionInParent());
					parentBranch.setChild(store,anotherBranch, n.getPositionInParent() + 1);
				}
				else {
					parentBranch.insertIntoNonLeaf(store,n.getKeys(n.getChildNumber() - 1), n, n.getPositionInParent());
					parentBranch.setChild(store,anotherBranch, n.getPositionInParent() + 1);
					//FIXME:This should be completed by a method
					n.setChild(store,null, n.getChildNumber());
					n.setKeys(null, n.getChildNumber() - 1);
					n.keyNumber--;
				}
				
				anotherBranch.setParent(store,parentBranch,n.getPositionInParent() + 1);
				
				n = parentBranch;
			}
		} finally {
            store.writeUnlock();
            writeLock.unlock();
        }
	}
	
    public void delete(Session session, PersistentStore store, Row row) {
       
    }
    void delete(PersistentStore store, NodeBTree x){
        
    }
    int find(Session session, NodeBTree root, Object[] rowdata){
    	int num=((NodeBTreeBranch)root).getChildNumber();
    	for(int i=0; i < num; i++){
    		Object[] Keys=((NodeBTreeBranch)root).getKeys(i);
    		if(compareKeys(session,rowdata,Keys) >= 0){
    			return i;
    		}
    	}
    	return num;
    }
	/**
     * Finds a match with a row from a different table
     *
     * @param session Session
     * @param store PersistentStore
     * @param rowdata array containing data for the index columns
     * @param rowColMap map of the data to columns
     * @param fieldCount int,number of columns of match
     * @param compareType int
     * @param readMode int
     * @return matching node or null
     */
    NodeBTreeData findNode(Session session, PersistentStore store, Object[] rowdata,
                     int[] rowColMap, int fieldCount, int compareType,
                     int readMode, boolean reversed) {

        readLock.lock();

        try {
            NodeBTreeBranch curNode = (NodeBTreeBranch)getAccessor(store),nextNode = null;
            NodeBTreeData result = null;
            Row     currentRow = null;

            /**
             * Why?
             */
            if (compareType != OpTypes.EQUAL
                    && compareType != OpTypes.IS_NULL) {
                fieldCount--;
            }
            
            while(!curNode.isLeaf){
            	
            	int i=find(session,curNode,rowdata);
            	curNode=(NodeBTreeBranch)curNode.getChild(store, i);
            }
            int i=find(session,curNode,rowdata);
            //result=(NodeBTreeData)curNode.getChild(store,i);
            switch(compareType){
            case OpTypes.IS_NULL:
            case OpTypes.EQUAL:
            	if(compareKeys(session,rowdata,curNode.getKeys(i)) != 0){
            		result=null;
            	}
            	else{
            		result=(NodeBTreeData)curNode.getChild(store,i);
            	}
            	break;
            case OpTypes.NOT:
            case OpTypes.GREATER:
            	if(compareKeys(session,rowdata,curNode.getKeys(i)) != 0){
            		result=(NodeBTreeData)curNode.getChild(store,i);
            	}
            	else{
            		result=(NodeBTreeData)curNode.getChild(store,i);
            		result=(NodeBTreeData)result.getNext(store);
            	}
            	break;
            case OpTypes.GREATER_EQUAL:
            	result=(NodeBTreeData)curNode.getChild(store,i);
            	break;
            	
            case OpTypes.SMALLER:
            	result=(NodeBTreeData)curNode.getChild(store,i);
            	result=result.getPre(store);
            	break;
            case OpTypes.SMALLER_EQUAL:
            	if(compareKeys(session,rowdata,curNode.getKeys(i)) != 0){
                	result=(NodeBTreeData)curNode.getChild(store,i);
                	result=result.getPre(store);
            	}
            	else{
            		result=(NodeBTreeData)curNode.getChild(store,i);
            	}
            	break;
            }
            
            // MVCC 190
            if (session == null) {
                return result;
            }

            while (result != null) {
                currentRow = result.getRow(store);

                if (session.database.txManager.canRead(session, currentRow,
                                                       readMode, colIndex)) {
                    break;
                }

                result = reversed ? (NodeBTreeData)last(store, result)
                                  : (NodeBTreeData)next(store, result);

                if (result == null) {
                    break;
                }

                currentRow = result.getRow(store);

                if (fieldCount > 0
                        && compareRowNonUnique(
                            session, currentRow.getData(), rowdata, rowColMap,
                            fieldCount) != 0) {
                    result = null;

                    break;
                }
            }

            return result;
        } finally {
            readLock.unlock();
        }
    }
}
